#include <fcntl.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <sys/time.h>
#include <sys/signal.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/msg.h>
#include <sys/ipc.h>
#include <unistd.h>

#include "../include/MQserver.h"
#include "../include/queries.h"
#include "../databaseADT/databaseADT.h"
#include "../../common/include/commonMQ.h"
#include "../../common/include/general.h"
#include "../../common/include/common.h"
#include "../../common/include/split.h"
#include "../include/server.h"

#define QUEUE_ID    200
int mqid;

/* Externs de client.c */
extern char splitVal;
extern splitNode splits[MAX_SPLITS];
extern int doLC;

int
mainMQ (void)
{
    int i, splitRet;


    if ((i = getOnlineFreeID(NULL)) == -1)
    {
        fatal(  "ERROR: Can't get a free slot in split vec.\n"
                "SOLUTION: Check MAX_SPLITS.\n");
        return -1;
    }

    /* Handlers de SIGS */
    (void)signal(SIGINT, sighandLCMQ);

    /* Actualizo online */
    modifyOnline(i, i, NULL, 1);

    /*Se splitea dependiendo de los parámetros.*/
    if (splitVal == 't')
        splitRet = split(listenPubMQ, NULL, (void *)&splits[i].thread, splitVal,
                endCiclesMQ);
    else
        splitRet = split(listenPubMQ, NULL, (void *)&splits[i].cID, splitVal,
                endCiclesMQ);

    if ( splitRet != 1)
        return fatal ("Split failed\n");

    readCommandsMQ();

    return 0;
}

void * listenPubMQ (void * paramsMQ)
{
    char msgbuf[MSGSIZE+1]= {0};
    struct msqid_ds mqstat;
    int move, cliID, splitRet, flag = 0;
    char *ptr = NULL;

    /* Creo la Cola de mensajes */
    mqid = msgget((key_t)QUEUE_ID, 0666 | IPC_CREAT);

    /* Get MQ status info. */
    if (msgctl (mqid, IPC_STAT, &mqstat) == -1)
    {
        fatal("MQ failed\n");
        return (void *)NULL;
    }

    while(doLC)
    {
        clrtext(msgbuf, MSGSIZE+1);
        flag = rcv_data_MQbl (mqid, PUBLIC_MQSERV, msgbuf);

        if(flag < 0)
            doLC = 0;
        else if (flag > 0)
        {
            move = getNumMQ(msgbuf, &cliID);
            ptr = strdup(msgbuf);

            /* HACK: Si id=0 no esta loggeado, no puedo splittear
                    Entonces llamo a waitPck sin splittear */
            if (cliID == 0)
                waitPckMQ((void *)ptr);

            else if (!isFreeOnline(cliID))
            {
                sndMQAns(mqid, cliID, IS_BUSY);
            }

            /* Me fijo si es un commit. Si es, creo el .new para evitar conflictos*/
            else if (canCommittMQ(msgbuf))
            {
                /* Actualizo online */
                modifyOnline(cliID, cliID, NULL, 1);

                /*Se splitea dependiendo de los parámetros.*/
                if (splitVal == 't')
                    splitRet = split(waitPckMQ, (void *)ptr,
                            (void *)&splits[cliID].thread, splitVal, NULL);
                else
                    splitRet = split(waitPckMQ, (void *)ptr,
                            (void *)&splits[cliID].cID, splitVal, NULL);

                if ( splitRet != 1)
                {
                    fatal(  "ERROR: listenPubPI Split failed.\n"
                            "SOLUTION: Check MAX_SPLITS.\n");
                    doLC = 0;
                }

            }
        }
    }
    if (msgctl (mqid, IPC_RMID, &mqstat) == -1)
    {
        fatal("MQ failed\n");
        return (void *)NULL;
    }

    return (void *)NULL;
}

void readCommandsMQ(void)
{
    int i;
    int cant_comms = 7;
    char command[MAX_COMMAND_LEN+1] = {0};
    char * param;
    char * commands[] = {"registrar", "usuarios", "dir", "last10",
                        "Ulast10", "newDir", "salir"};

    void (*pFunct[]) (char *)= {registrar, usuarios, dir, last10, ulast10,
                                newDir, salirMQ};

    fprintf (stdout, "\n<>: ");
    getLine (command, MAX_COMMAND_LEN, stdin);

    for (i=0 ; i < cant_comms+1 ; i++)
    {
        if (i == cant_comms)
        {
           fprintf (stdout, "\nError: Invalid command\n<>: ");
           getLine (command, MAX_COMMAND_LEN, stdin);
           i = -1;
        }
        else if (strcmp (command, commands[i]) == 0)
        {
            if(i == 6)
                salirMQ(NULL);

            pFunct[i] (param);
            fprintf (stdout, "\n<>: ");
            getLine (command, MAX_COMMAND_LEN, stdin);
            i = -1;
        }
    }

    return ;
}

void * waitPckMQ (void * ptr)
{
    int com;
    int size;
    char params[50] = {0};
    char priv[50] = {0};
    char full[100] = {0};
    char aux_msg[MSGSIZE+1] = {0};
    int cliID, privID;
    char *msg = (char *)ptr;
    int close = 1;

    sscanf(msg, "%d,%d,%s", &privID, &com, aux_msg);

    if (privID == 0)
        close = 0;

    if (com == COM_NAME)
        sscanf(aux_msg, "%d,%d,%s", &cliID, &size,full);
    else
        sscanf(aux_msg, "%d,%s", &size,full);

    full[size] = 0;

    getparamsMQ(full, params, priv);

    switch(com)
    {
        case COM_NAME:
            (void)addnameMQ(cliID, priv);
            break;

        case COM_LIST:
            listdirsMQ(full, privID);
            break;

        case COM_ADD:
            (void)adddirMQ(params, priv);
            break;

        case COM_REMOVE:
            (void)removedirMQ(params, priv);
            break;

        case COM_CLOSE:
            closesessMQ(full);
            break;

        case COM_COMMIT:
            rcvCommitMQ(params, priv);
            break;

        default:
            break;
    }
    free(ptr);

    if (close)
    {
        modifyOnline(privID, privID, NULL, 0);
        exitSplit(splitVal);
    }
    return 0;
}


void salirMQ(char * param)
{
    closeAllMQ();
    return;
}

int addnameMQ (int cliID, char * name)
{
    char priv[MAX_NAME_LEN] = {0};
    char out_msg[MSGSIZE+1] = {0};
    int size, pubcliID;

    strcpy(priv, name);

    if (addUser(name, priv) == DB_INTERNAL_ERROR)
    {
        fprintf(stderr, "DataBase error\n");
        sndMQAns(mqid, cliID, COM_ERR);
        return -1;
    }

    switch(DBconnect(name))
    {
        case DB_INTERNAL_ERROR:
            sndMQAns(mqid, cliID, COM_ERR);
            return fatal(   "ERROR: Database error In addname.\n"
                            "SOLUTION: Check Database status.\n");
            break;

        case DB_ALREADY_EXISTS:
            sndMQAns(mqid, cliID, COM_ERR);
            fprintf(stderr, "ERROR: User %s already logged\n", name);
            return fatal(   "SOLUTION: If this persist, ban him. :)\n");
            break;

        case DB_NO_SLOTS:
            sndMQAns(mqid, cliID, COM_ERR);
            return fatal(   "ERROR: No more slots available.\n"
                            "SOLUTION: Check MAX_SPLITS.\n");
            return -1;
            break;

        default:
            break;
    }


    pubcliID = userID(priv);

    sprintf(out_msg, "%d,%d", COM_ACK, pubcliID);
    size = strlen(out_msg);
    send_data_MQ (mqid, cliID, out_msg, size);

    addLog(priv, "Logged in");

    return 1;

}

void adddirMQ (char * dir, char * priv)
{
    char answ[50] = {0};
    int resp, back, len;
    char path[MAX_DIR_LEN + 7] = {0};
    char logact[100] = {0};
    int cliID, privCli, servID, move;
    char aux_msg[MSGSIZE+1];

    cliID = userID(priv);
    servID = cliID * 10;

    queueADT dirs;
    int qlen, flag = 1;
    char *aux_dir;

    if(isSynched(cliID, dir))
    {
        sndMQAns(mqid, cliID, IS_SYNC);
        return;
    }

    dirs = listDir();
    qlen = queueLength(dirs);

    while (qlen-- && flag)
    {
        aux_dir = (char *)dequeue ( dirs );

        if (strcmp(dir, aux_dir) == 0)
            flag = 0;

        free(aux_dir);
    }

    if (flag)
    {
        sndMQAns(mqid, cliID, DIR_ERR);
        return;
    }

    len = strlen((char *)dir);
    sprintf(answ, "%d,%d,%d,%s", COM_SND_FILES, servID, len, dir);
    len = strlen((char *)answ);
    send_data_MQ (mqid, cliID, answ, len);

    if (rcv_data_MQbl (mqid, servID, aux_msg)< 0)
    {
        fatal ("ERROR: Msg read failed\n"
                "SOLUTION: Check server's addirMQ\n");
        return ;
    }

    move = getNumMQ (aux_msg, &back);
    move += getNumMQ (aux_msg + move, &privCli);

    resp = addFifoDir(priv,dir);
    sprintf (path, "./%s", dir);
    if (walkDirMQ(path, "", mqid, privCli, servID) <= 0)
    {
        printf ("ERROR: Dir send failed\n"
                "SOLUTION: Check dir: %s\n", path);
        return ;
    }

    sndMQAns(mqid, privCli, COM_END);

    strcpy (logact, "Sync dir: ");
    strcat(logact, dir);
    addLog(priv, logact);

    return;
}

void removedirMQ (char * dir, char * priv)
{
    DB_ERR ans;
    char logact[100] = {0};
    int len;
    char answ[MSGSIZE] = {0};
    int servID;
    int cliID;

    cliID = userID(priv);
    servID = cliID * 10;


    if ((ans = removeDirByFifo(priv, dir)) != DB_SUCCESS)
    {
        sndMQAns(mqid, cliID, COM_ERR);
        fatal ("ERROR: Database error In removedirMQ.\n"
                "SOLUTION: Check Database status.\n");
        return ;
    }

    len = strlen(dir);
    sprintf(answ, "%d,%d,%d,%s", COM_RMV_OK, servID, len, dir);
    len = strlen(answ);

    send_data_MQ (mqid, cliID, answ, len);

    strcpy (logact, "Remove sync: ");
    strcat(logact, dir);
    addLog(priv, logact);

    return ;

    return;
}

void listdirsMQ (char * priv, int cliID)
{
    char answ[50] = {0};
    int servID, ret = 0;
    int privCli, len, move, comm;
    char aux_msg[MSGSIZE+1] = {0};

//    cliID = userID(priv);
    servID = cliID * 10;

    /* Creo el MQtype privado del server */
    sprintf(answ, "%d,%d", COM_SND_DIRLIST, servID);

    len = strlen((char *)answ);

    if ((ret = send_data_MQ (mqid, cliID, answ, len)) < 0)
    {
        fatal ("Error: send msg failed\n"
                "SOLUTION: Check server's listdirsMQ\n");

    }

    if (rcv_data_MQbl (mqid, servID, aux_msg)< 0)
    {
        fatal ("ERROR: msg read failed\n"
                "SOLUTION: Check server's listdirsMQ\n");
        return ;
    }

    move = getNumMQ (aux_msg, &comm);
    move += getNumMQ (aux_msg + move, &privCli);


    if (sndlistMQ (mqid, servID, privCli) < 0)
    {
        fatal ("ERROR: List could not be sent correctly\n"
                "SOLUTION: Check server's sndlistMQ\n");
        return ;
    }

    addLog(priv, "List dirs");

    return;
}

void closesessMQ (char * priv)
{
    DB_ERR ans;
    int cliID;

    cliID = userID(priv);

    if ((ans = disconnectFifo(priv)) != DB_SUCCESS)
    {
        sndMQAns(mqid, cliID, COM_ERR);
        fatal ("ERROR: Database error In closessMQ.\n"
                "SOLUTION: Check Database status.\n");
    }

    addLog(priv, "Logged Out");

    return;
}

void getparamsMQ (char * full, char * params, char * priv)
{
    char c;

    c = *full++;
    while (c != ',' && c != 0)
    {
        *priv++ = c;
        c = *full++;
    }
    if (c == 0)
        return ;

    strcpy(params, full);
    return;
}

int sndlistMQ (int mqid, int servID, int cliID)
{
    queueADT dirs;
    int name_len, tot_len, cant, back;
    char answ[MSGSIZE + 1] = {0};
    char * name;

    dirs = listDir();
    cant = queueLength(dirs);

    while (cant--)
    {
        name = (char *)dequeue ( dirs );
        name_len = strlen( name);
        sprintf(answ, "%d,%d,%s", SND_DIR_NAME, name_len, name);

        tot_len = strlen (answ);

        do
        {
            if ((send_data_MQ (mqid, cliID, answ, tot_len)) < 0)
            {
                fatal ("ERROR: send dir failed\n"
                        "SOLUTION: check server's sndlistMQ\n");
                return -1;
            }


            back = rcvMQAns(mqid, servID);

        }while (back != COM_ACK);
    }


    freeQueue( dirs );
    sndMQAns(mqid, cliID, COM_END);

    return 1;
}

void rcvCommitMQ (char * params, char * priv)
{
    char answ[50] = {0};
    char buffer[BUF_SIZE] = {0};
    char newpath[100] = {0};
    FILE * pfile;
    int path_size, comm;
    int nread, len, size, type, move;
    int servID, cliID, privcliID;
    char path[100] = {0};
    char msgbuf[100] = {0};
    char logact[100] = {0};
    char dir[FILE_LEN] = {0};
    char aux_msg[MSGSIZE+1] = {0};

    cliID = userID(priv);
    servID = cliID * 10;

    getDirType(params, dir, &type);


    if (type == ADD_MOD)
    {
        len = strlen(dir);
        sprintf(answ, "%d,%d,%d,%s", COM_RCV_COMMIT, servID, len, dir);

        len = strlen((char *)answ);

        send_data_MQ (mqid, cliID, answ, len);

        if (rcv_data_MQbl (mqid, servID, aux_msg)< 0)
        {
            fatal ("ERROR: msg read failed\n"
                    "SOLUTION: Check server's rcvCommitMQ\n");
            return ;
        }

        move = getNumMQ (aux_msg, &comm);

        if (comm != SND_PRIV)
        {
            printf("Error receiving priv\n");
            return;
        }
        move += readNumMQ (aux_msg + move, &privcliID);
        sndMQAns(mqid, privcliID, COM_ACK);

        clrtext(aux_msg, MSGSIZE+1);
        if (rcv_data_MQbl (mqid, servID, aux_msg)< 0)
        {
            fatal ("msg read failed\n"
                    "SOLUTION: Check server's rcvCommitMQ\n");
            return ;
        }

        move = getNumMQ (aux_msg, &comm);

        if (comm == SND_FILE)
        {
            move += getNumMQ (aux_msg + move, &path_size);

            sprintf(msgbuf, "%s", aux_msg + move);
            msgbuf[path_size] = 0;
            sscanf (msgbuf, "%d,%s", &size, path);
            sprintf(newpath, "%s.new", path);

            if (size == -1)
            {
                if(mkdir(path, 0777) == -1)
                {
                    printf("ERROR: creating dir: %s\n", path);
                    sndMQAns(mqid, privcliID, COM_ERR);
                    return ;
                }
                if (remove(newpath) != 0)
                {
                    printf("ERROR: removing dir: %s\n", newpath);
                    sndMQAns(mqid, privcliID, COM_ERR);
                    return ;
                }
                sndMQAns(mqid, privcliID, COM_ACK);
            }
            else
            {
                if((pfile = fopen(newpath, "w")) == NULL)
                {
                    fatal ("ERROR: creating file\n"
                            "SOLUTION: Check server's rcvCommitMQ\n");
                    return ;
                }

                nread = 0;
                sndMQAns(mqid, privcliID, COM_ACK);
                while (size > 0)
                {
                    if (size >= BUF_SIZE)
                        nread = BUF_SIZE;
                    else
                        nread = size;

                    if (nrcv_data_MQ (mqid, servID, buffer, nread) < 0)
                    {
                        sndMQAns(mqid, cliID, COM_ERR);
                        printf ("msg read failed\n");
                        return ;
                    }

                    if ((fwrite(buffer, 1, nread, pfile)) < nread)
                    {
                        sndMQAns(mqid, cliID, COM_ERR);
                        printf ("File write failed\n");
                        return ;
                    }

                    sndMQAns(mqid, privcliID, COM_ACK);

                    size -= nread;
                }

                if ((fclose (pfile)) == EOF)
                {
                    printf ("file close failed\n");
                    return ;
                }

                if (rename(newpath, path) != 0)
                {
                    printf ("Error renaming %s\n", newpath);
                    return;
                }
            }

        }

        if (srv_commitMQ (servID, cliID, path, type) < 0)
        {
            printf("Error forwarding %s\n", path);
            return ;
        }

        strcpy(logact, "Commited (add/mod): ");
        strcat (logact, path);

        addLog(priv, logact);
        sndMQAns(mqid, privcliID, COM_ACK);

    }
    else if(type == DEL_COMMIT)
    {
        clrtext(aux_msg, MSGSIZE+1);
        if (remove(dir) != 0)
        {
            sndMQAns(mqid, cliID, COM_ERR);
        }
        else
        {
            if (srv_commitMQ (servID, cliID, dir, type) < 0)
            {
                printf("Error forwarding %s\n", path);
                return ;
            }
            strcpy(logact, "Commited (del): ");
            strcat (logact, dir);

            addLog(priv, logact);
            sndMQAns(mqid, cliID, COMMIT_OK);
        }
    }
    else if(type == DEL_DIR_COMMIT)
    {
        if (rmdir(dir) != 0)
        {
            sndMQAns(mqid, cliID, COM_ERR);
        }
        else
        {
            if (srv_commitMQ(servID, cliID, dir, type) < 0)
            {
                printf("Error forwarding %s\n", path);
                return;
            }

            strcpy(logact, "Commited (dir del): ");
            strcat (logact, dir);

            addLog(priv, logact);

            sndMQAns(mqid, cliID, COMMIT_OK);
        }

    }

    return;
}

int srv_commitMQ (int servID, int cliID, char * path, int type)
{
    queueADT clients;
    int cant;
    int back;
    char dir[MAX_DIR_LEN] = {0};
    int len;
    char answ[100] = {0};
    char aux_msg[MSGSIZE+1] = {0};
    int * next_cliID;
    int next_servID;
    int move, privcliID;

    getDir (dir, path);
    clients = updateIDs(cliID, dir);
    cant = queueLength(clients);

    while (cant--)
    {
        next_cliID = (int *)dequeue ( clients );
        next_servID = (*next_cliID) * 10;
        while (!isFreeOnline(*next_cliID))
            sleep(5);
        sprintf(aux_msg, "%s,%d", path, type);
        len = strlen((char *)aux_msg);
        sprintf(answ, "%d,%d,%d,%s", COM_SND_COMMIT, next_servID, len, aux_msg);

        len = strlen((char *)answ);

        clrtext(aux_msg, MSGSIZE+1);
        send_data_MQ (mqid, *next_cliID, answ, len);

        if (rcv_data_MQbl (mqid, next_servID, aux_msg)< 0)
            fatal ("msg read failed\n");

        move = getNumMQ(aux_msg, &back);
        move += readNumMQ(aux_msg + move, &privcliID);

        if (type == ADD_MOD)
        {
            if (snd_file_MQ (mqid, privcliID, next_servID, path, 0) < 0)
            {
                printf ("%s not sent to %d\n", path, privcliID);
            }
        }

    }

    freeQueue( clients );

    return 1;
}

int canCommittMQ(char * msg)
{
    FILE * pfile;
    int com;
    int size;
    char params[50] = {0};
    char priv[50] = {0};
    char full[100] = {0};
    int type;
    char dir[FILE_LEN] = {0};
    char newpath[100] = {0};
    int len;
    char answ[FILE_LEN] = {0};
    int cliID;
    char aux_msg[MSGSIZE+1] = {0};

    sscanf(msg, "%d,%d,%s", &cliID, &com, aux_msg);

    if (com == COM_NAME)
        return 1;
    else
        sscanf(aux_msg, "%d,%s", &size,full);

    if (com != COM_COMMIT)
        return 1;

    full[size] = 0;

    getparamsMQ(full, params, priv);
    getDirType(params, dir, &type);

    sprintf(newpath, "%s.new", dir);

    if((pfile = fopen(newpath, "r")) == NULL)
    {
        if (type == DEL_COMMIT)
            return 1;
        if (type == DEL_DIR_COMMIT)
            return 1;
        if((pfile = fopen(newpath, "w")) == NULL)
        {
            fatal ("file open failed\n");
            return -1;
        }

        if (fclose(pfile) == EOF)
        {
            fatal ("file close failed\n");
            return -1;
        }
        return 1;

    }

    len = strlen(dir);
    sprintf(answ, "%d,%d,%s", WILL_UPDATE, len, dir);

    len = strlen((char *)answ);
    send_data_MQ (mqid, cliID, answ, len);

    return 0;
}
void endCiclesMQ (int signo)
{
    doLC = 0;
    return ;
}

void closeAllMQ(void)
{
    queueADT q = NULL;
    int *index = NULL;
    int stat;

    printf("Closing splits.");

    doLC = 0;
    if (splitVal == 't')
    {
        q = getUsedOnline();
        while((index = dequeue(q)) != NULL)
        {
            pthread_join(splits[*index].thread, NULL);
             printf(".");
            free(index);
        }
        freeQueue(q);
    }
    else
    {
        /*HACK: Mando senial para salir del select*/
        kill(splits[0].cID, SIGUSR1);
        sleep(1);
        while ((waitpid (-1, &stat, 0)) > 0)
            printf(".");
    }

    printf("\nServer closed\n");
    exitSplit(splitVal);
}
void
sighandLCMQ(int signo)
{
    closeAllMQ();
}

int
rcv_data_MQbl (int mqid, int mqtype, char * r_msg)
{
    struct mymsg msg;
    int rtrn = 0;
    int i = 2;

    msg.mtype = mqtype;
    clrtext(msg.mtext, FILE_LEN);

    errno = ENOMSG;

    do
    {
        if ((rtrn = msgrcv(mqid, &msg, MSGSIZE, mqtype, IPC_NOWAIT)) == -1)
        {
            if(errno != ENOMSG)
            {
                printf("msgop: msgrcv failed");
                return -1;
            }
        }
        else
        {
            sprintf(r_msg, "%s", msg.mtext);
            return rtrn;
        }
        sleep(SEC_TIMEOUT);
    } while (i--);
    return 0;
}
